package apache

import (
	"context"
	"sync"
	"time"

	"github.com/pkg/errors"

	"github.com/apache/rocketmq-client-go/v2/consumer"

	"github.com/apache/rocketmq-client-go/v2"
	"github.com/apache/rocketmq-client-go/v2/primitive"
	"github.com/apache/rocketmq-client-go/v2/producer"
	"testkjcld.bhecard.com/gitlab/kit/esim/config"
	"testkjcld.bhecard.com/gitlab/kit/esim/log"
	"testkjcld.bhecard.com/gitlab/kit/esim/rocketmq/internal"
)

var (
	poolOnce   sync.Once
	onceClient *MQClient
)

const (
	SendSync = iota
	SendAsync
)

type MQClientOptions struct{}

type Option func(*MQClient)

func (MQClientOptions) WithConf(conf config.Config) Option {
	return func(mq *MQClient) {
		mq.conf = conf
	}
}

func (MQClientOptions) WithLogger(logger log.Logger) Option {
	return func(mq *MQClient) {
		mq.logger = logger
	}
}

type MQClient struct {
	logger log.Logger

	conf config.Config

	endpoint []string // rocketMQ URL

	groupName string // 消费组

	nameSpace string // 命名空间

	retryCount int // 重试次数

	apacheCredentials primitive.Credentials

	producer rocketmq.Producer // apache producer

	mu          sync.Mutex
	consumerMap map[string]*ApacheMQConsumer // 消费者实例

	transProducerMap sync.Map // 事务实例

	listener sync.Map // 注册时间

	syncMode int
}

func NewMQClient(options ...Option) *MQClient {
	poolOnce.Do(func() {
		var err error
		onceClient = &MQClient{}
		for _, option := range options {
			option(onceClient)
		}

		if onceClient.conf == nil {
			onceClient.conf = config.NewNullConfig()
		}

		if onceClient.logger == nil {
			onceClient.logger = log.NewLogger()
		}

		onceClient.consumerMap = make(map[string]*ApacheMQConsumer)

		// 配置数据
		onceClient.endpoint = onceClient.conf.GetStringSlice("apache_endpoint")
		if onceClient.endpoint == nil {
			onceClient.logger.Panicf("apache_endpoint is null! please confirm")
		}

		// 配置数据
		onceClient.groupName = onceClient.conf.GetString("apache_group_name")

		onceClient.nameSpace = onceClient.conf.GetString("apache_name_space")

		onceClient.retryCount = onceClient.conf.GetInt("apache_retry_count")
		if onceClient.retryCount == 0 {
			onceClient.retryCount = 2
		}

		// 消息发送模式
		syncMode := onceClient.conf.GetString("apache_sync_mode")
		if syncMode == "async" {
			onceClient.syncMode = SendAsync
		}

		// 配置数据
		accessKey := onceClient.conf.GetString("apache_access_key")
		secret := onceClient.conf.GetString("apache_secret")
		if accessKey != "" && secret != "" {
			onceClient.apacheCredentials = primitive.Credentials{
				AccessKey:     accessKey,
				SecretKey:     secret,
				SecurityToken: "",
			}
		}
		// 生产者
		onceClient.producer, err = rocketmq.NewProducer(
			producer.WithNsResolver(primitive.NewPassthroughResolver(onceClient.endpoint)),
			producer.WithRetry(onceClient.retryCount),
			producer.WithCredentials(onceClient.apacheCredentials),
			producer.WithNamespace(onceClient.nameSpace),
		)
		if err != nil {
			onceClient.logger.Panicf("[rocket mq] producer init fail %v", err)
		}

		err = onceClient.producer.Start()
		if err != nil {
			onceClient.logger.Panicf("[rocket mq] producer start fail %v", err)
		}

		onceClient.logger.Infof("[rocket mq] init success %s",
			onceClient.endpoint)
	})

	return onceClient
}

func (mc *MQClient) CreateConsumer(topicName, groupID, messageTag string) (internal.MQConsumer, error) {
	// 如果groupId为空，则每次不再配置新的Consumer
	if groupID == "" {
		// 创建apache  消费者
		c, err := rocketmq.NewPushConsumer(
			consumer.WithNsResolver(primitive.NewPassthroughResolver(onceClient.endpoint)),
			consumer.WithGroupName(groupID),
			consumer.WithRetry(onceClient.retryCount),
			consumer.WithCredentials(onceClient.apacheCredentials),
			consumer.WithNamespace(onceClient.nameSpace),
		)
		if err != nil {
			mc.logger.Errorf("[rocket mq] consumer init fail %v", err)
			return nil, err
		}
		return &ApacheMQConsumer{
			PushConsumer: c,
		}, nil
	}

	key := groupID
	if v, ok := mc.consumerMap[key]; ok {
		return v, nil
	}

	mc.mu.Lock()
	defer mc.mu.Unlock()

	if v, ok := mc.consumerMap[key]; ok {
		return v, nil
	}

	// 创建apache  消费者
	c, err := rocketmq.NewPushConsumer(
		consumer.WithNsResolver(primitive.NewPassthroughResolver(onceClient.endpoint)),
		consumer.WithGroupName(groupID),
		consumer.WithRetry(onceClient.retryCount),
		consumer.WithCredentials(onceClient.apacheCredentials),
		consumer.WithNamespace(onceClient.nameSpace),
	)
	if err != nil {
		mc.logger.Errorf("[rocket mq] consumer init fail %v", err)
		return nil, err
	}
	wrapC := &ApacheMQConsumer{
		PushConsumer: c,
	}
	mc.consumerMap[key] = wrapC
	return wrapC, nil
}

func (mc *MQClient) PublishMessage(topicName, messageBody string) error {
	msg := &primitive.Message{
		Topic: topicName,
		Body:  []byte(messageBody),
	}
	return mc.publish(topicName, msg)
}

// PublishMsgWithTag with message tag and key.
func (mc *MQClient) PublishMsgWithTag(topicName, messageBody, messageTag string) error {
	msg := &primitive.Message{
		Topic: topicName,
		Body:  []byte(messageBody),
	}
	msg.WithTag(messageTag)
	return mc.publish(topicName, msg)
}

// PublishMsgWithKeyTag with message tag.
func (mc *MQClient) PublishMsgWithKeyTag(topicName, messageBody, messageTag, messageKey string) error {
	msg := &primitive.Message{
		Topic: topicName,
		Body:  []byte(messageBody),
	}
	msg.WithTag(messageTag)
	msg.WithKeys([]string{messageKey})
	return mc.publish(topicName, msg)
}

func (mc *MQClient) PublishDelayMessage(topicName, messageBody string, delay time.Duration) error {
	msg := &primitive.Message{
		Topic: topicName,
		Body:  []byte(messageBody),
	}
	msg.WithDelayTimeLevel(int(delay / time.Second))
	return mc.publish(topicName, msg)
}

func (mc *MQClient) PublishDelayMsgWithTag(topicName, messageBody, messageTag string, delay time.Duration) error {
	msg := &primitive.Message{
		Topic: topicName,
		Body:  []byte(messageBody),
	}
	msg.WithTag(messageTag)
	msg.WithDelayTimeLevel(int(delay / time.Second))
	return mc.publish(topicName, msg)
}

func (mc *MQClient) PublishDelayMsgWithKeyTag(topicName, messageBody, messageTag, messageKey string, delay time.Duration) error {
	msg := &primitive.Message{
		Topic: topicName,
		Body:  []byte(messageBody),
	}
	msg.WithTag(messageTag)
	msg.WithKeys([]string{messageKey})
	msg.WithDelayTimeLevel(int(delay / time.Second))
	return mc.publish(topicName, msg)
}

func (mc *MQClient) PublishMessageProp(topicName, messageTag, messageBody string, properties map[string]string) error {
	msg := &primitive.Message{
		Topic: topicName,
		Body:  []byte(messageBody),
	}
	msg.WithTag(messageTag)
	msg.WithProperties(properties)
	return mc.publish(topicName, msg)
}

func (mc *MQClient) PublishDelayMessageProt(topicName, messageBody string, properties map[string]string, delay time.Duration) error {
	msg := &primitive.Message{
		Topic: topicName,
		Body:  []byte(messageBody),
	}
	msg.WithDelayTimeLevel(int(delay / time.Second))
	msg.WithProperties(properties)
	return mc.publish(topicName, msg)
}

func (mc *MQClient) publish(topicName string, msg *primitive.Message) error {
	var err error
	var resp *primitive.SendResult
	if mc.syncMode == SendSync {
		resp, err = mc.producer.SendSync(context.Background(), msg)
		if err != nil {
			mc.logger.Errorf("发送信息失败[%s]:[%s]", topicName, err)
			return err
		}
		mc.logger.Infof("Publish成功 ---->MessageId:[%s][%d];", resp.MsgID, resp.Status)
	} else {
		err = mc.producer.SendAsync(context.Background(), func(ctx context.Context, result *primitive.SendResult, err error) {
			if err != nil {
				mc.logger.Errorc(ctx, "SendAsync发送信息失败[%s]:[%s]", topicName, err)
				return
			}
			mc.logger.Errorc(ctx, "SendAsync发送信息结果[%s]:[%s]-[%d]", topicName, result.MsgID, result.Status)
		}, msg)
		if err != nil {
			mc.logger.Errorf("SendAsync发送信息失败[%s]:[%s]", topicName, err)
			return err
		}
	}
	return nil
}

func (mc *MQClient) RegisterListener(topicName string, listener primitive.TransactionListener) {
	if _, ok := mc.listener.Load(topicName); ok {
		mc.logger.Panicf("注册事务监听处理器失败,[%s]已存在;", topicName)
	}
	mc.listener.LoadOrStore(topicName, listener)
}

func (mc *MQClient) tranProducer(topicName, groupID string) (internal.MQTransProducer, error) {
	lis, ok := mc.listener.Load(topicName)
	if !ok {
		return nil, errors.Errorf("topic[%s]未注册;", topicName)
	}
	// 创建apache transaction 生产者
	p, _ := rocketmq.NewTransactionProducer(
		lis.(primitive.TransactionListener),
		producer.WithNsResolver(primitive.NewPassthroughResolver(onceClient.endpoint)),
		producer.WithGroupName(groupID),
		producer.WithRetry(onceClient.retryCount),
	)
	err := p.Start()
	if err != nil {
		mc.logger.Errorf("start transaction producer error: %v", err)
		return nil, err
	}
	return &ApacheMQTranProducer{
		TransactionProducer: p,
		TopicName:           topicName,
		GroupName:           groupID,
	}, nil
}

func (mc *MQClient) CreateTransProducer(topicName, groupID string) (internal.MQTransProducer, error) {
	p, ok := mc.transProducerMap.Load(topicName + groupID)
	if ok {
		return p.(*ApacheMQTranProducer), errors.Errorf("[%s][%s]TransProducer already create", topicName, groupID)
	}

	v, err := mc.tranProducer(topicName, groupID)
	if err != nil {
		mc.logger.Errorf("tranProducer 创建事务生产者失败[%s][%s] err[%v]", topicName, groupID, err)
		return v, err
	}

	p, _ = mc.transProducerMap.LoadOrStore(topicName+groupID, p)
	return p.(*ApacheMQTranProducer), nil
}

func (mc *MQClient) loadTransProducer(topicName, groupID string) (internal.MQTransProducer, error) {
	p, ok := mc.transProducerMap.Load(topicName + groupID)
	if ok {
		return p.(*ApacheMQTranProducer), nil
	}
	return nil, errors.Errorf("[%s] Transaction Producer not redister listen;", topicName)
}

// PublishWithCommit trans commit.
func (mc *MQClient) PublishWithCommit(topicName, groupID, handler string) error {
	return nil
}

// PublishWithRollBack trans rollback.
func (mc *MQClient) PublishWithRollBack(topicName, groupID, handler string) error {
	return nil
}

// PublishMsgWithTagTrans with message tag and group trans.
func (mc *MQClient) PublishMsgWithTagTrans(topicName, groupID, messageBody, messageTag string, tradeID map[string]string) (*internal.ConsumeMessageAck, error) {
	p, err := mc.loadTransProducer(topicName, groupID)
	if err != nil {
		mc.logger.Errorf("PublishMsgWithTagTrans loadTransProducer 失败[%s]", err)
		return nil, err
	}

	msg := &primitive.Message{
		Topic: topicName,
		Body:  []byte(messageBody),
	}
	msg.WithTag(messageTag)
	msg.WithProperties(tradeID)

	r, err := p.SendMessageInTransaction(context.Background(), msg)
	if err != nil {
		mc.logger.Errorf("SendMessageInTransaction 事务信息发送失败[%s][%s]:[%s]", topicName, messageTag, err)
		return nil, err
	}

	mc.logger.Infof("SendMessageInTransaction 事务信息发送成功 ---->MessageId:[%s], Status:[%d];", r.MsgID, r.Status)
	return &internal.ConsumeMessageAck{
		MessageID: r.MsgID,
	}, nil
}
